123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312 |
- """
- Tagged JSON
- ~~~~~~~~~~~
- A compact representation for lossless serialization of non-standard JSON
- types. :class:`~flask.sessions.SecureCookieSessionInterface` uses this
- to serialize the session data, but it may be useful in other places. It
- can be extended to support other types.
- .. autoclass:: TaggedJSONSerializer
- :members:
- .. autoclass:: JSONTag
- :members:
- Let's see an example that adds support for
- :class:`~collections.OrderedDict`. Dicts don't have an order in JSON, so
- to handle this we will dump the items as a list of ``[key, value]``
- pairs. Subclass :class:`JSONTag` and give it the new key ``' od'`` to
- identify the type. The session serializer processes dicts first, so
- insert the new tag at the front of the order since ``OrderedDict`` must
- be processed before ``dict``.
- .. code-block:: python
- from flask.json.tag import JSONTag
- class TagOrderedDict(JSONTag):
- __slots__ = ('serializer',)
- key = ' od'
- def check(self, value):
- return isinstance(value, OrderedDict)
- def to_json(self, value):
- return [[k, self.serializer.tag(v)] for k, v in iteritems(value)]
- def to_python(self, value):
- return OrderedDict(value)
- app.session_interface.serializer.register(TagOrderedDict, index=0)
- """
- import typing as t
- from base64 import b64decode
- from base64 import b64encode
- from datetime import datetime
- from uuid import UUID
- from markupsafe import Markup
- from werkzeug.http import http_date
- from werkzeug.http import parse_date
- from ..json import dumps
- from ..json import loads
- class JSONTag:
- """Base class for defining type tags for :class:`TaggedJSONSerializer`."""
- __slots__ = ("serializer",)
- #: The tag to mark the serialized object with. If ``None``, this tag is
- #: only used as an intermediate step during tagging.
- key: t.Optional[str] = None
- def __init__(self, serializer: "TaggedJSONSerializer") -> None:
- """Create a tagger for the given serializer."""
- self.serializer = serializer
- def check(self, value: t.Any) -> bool:
- """Check if the given value should be tagged by this tag."""
- raise NotImplementedError
- def to_json(self, value: t.Any) -> t.Any:
- """Convert the Python object to an object that is a valid JSON type.
- The tag will be added later."""
- raise NotImplementedError
- def to_python(self, value: t.Any) -> t.Any:
- """Convert the JSON representation back to the correct type. The tag
- will already be removed."""
- raise NotImplementedError
- def tag(self, value: t.Any) -> t.Any:
- """Convert the value to a valid JSON type and add the tag structure
- around it."""
- return {self.key: self.to_json(value)}
- class TagDict(JSONTag):
- """Tag for 1-item dicts whose only key matches a registered tag.
- Internally, the dict key is suffixed with `__`, and the suffix is removed
- when deserializing.
- """
- __slots__ = ()
- key = " di"
- def check(self, value: t.Any) -> bool:
- return (
- isinstance(value, dict)
- and len(value) == 1
- and next(iter(value)) in self.serializer.tags
- )
- def to_json(self, value: t.Any) -> t.Any:
- key = next(iter(value))
- return {f"{key}__": self.serializer.tag(value[key])}
- def to_python(self, value: t.Any) -> t.Any:
- key = next(iter(value))
- return {key[:-2]: value[key]}
- class PassDict(JSONTag):
- __slots__ = ()
- def check(self, value: t.Any) -> bool:
- return isinstance(value, dict)
- def to_json(self, value: t.Any) -> t.Any:
- # JSON objects may only have string keys, so don't bother tagging the
- # key here.
- return {k: self.serializer.tag(v) for k, v in value.items()}
- tag = to_json
- class TagTuple(JSONTag):
- __slots__ = ()
- key = " t"
- def check(self, value: t.Any) -> bool:
- return isinstance(value, tuple)
- def to_json(self, value: t.Any) -> t.Any:
- return [self.serializer.tag(item) for item in value]
- def to_python(self, value: t.Any) -> t.Any:
- return tuple(value)
- class PassList(JSONTag):
- __slots__ = ()
- def check(self, value: t.Any) -> bool:
- return isinstance(value, list)
- def to_json(self, value: t.Any) -> t.Any:
- return [self.serializer.tag(item) for item in value]
- tag = to_json
- class TagBytes(JSONTag):
- __slots__ = ()
- key = " b"
- def check(self, value: t.Any) -> bool:
- return isinstance(value, bytes)
- def to_json(self, value: t.Any) -> t.Any:
- return b64encode(value).decode("ascii")
- def to_python(self, value: t.Any) -> t.Any:
- return b64decode(value)
- class TagMarkup(JSONTag):
- """Serialize anything matching the :class:`~markupsafe.Markup` API by
- having a ``__html__`` method to the result of that method. Always
- deserializes to an instance of :class:`~markupsafe.Markup`."""
- __slots__ = ()
- key = " m"
- def check(self, value: t.Any) -> bool:
- return callable(getattr(value, "__html__", None))
- def to_json(self, value: t.Any) -> t.Any:
- return str(value.__html__())
- def to_python(self, value: t.Any) -> t.Any:
- return Markup(value)
- class TagUUID(JSONTag):
- __slots__ = ()
- key = " u"
- def check(self, value: t.Any) -> bool:
- return isinstance(value, UUID)
- def to_json(self, value: t.Any) -> t.Any:
- return value.hex
- def to_python(self, value: t.Any) -> t.Any:
- return UUID(value)
- class TagDateTime(JSONTag):
- __slots__ = ()
- key = " d"
- def check(self, value: t.Any) -> bool:
- return isinstance(value, datetime)
- def to_json(self, value: t.Any) -> t.Any:
- return http_date(value)
- def to_python(self, value: t.Any) -> t.Any:
- return parse_date(value)
- class TaggedJSONSerializer:
- """Serializer that uses a tag system to compactly represent objects that
- are not JSON types. Passed as the intermediate serializer to
- :class:`itsdangerous.Serializer`.
- The following extra types are supported:
- * :class:`dict`
- * :class:`tuple`
- * :class:`bytes`
- * :class:`~markupsafe.Markup`
- * :class:`~uuid.UUID`
- * :class:`~datetime.datetime`
- """
- __slots__ = ("tags", "order")
- #: Tag classes to bind when creating the serializer. Other tags can be
- #: added later using :meth:`~register`.
- default_tags = [
- TagDict,
- PassDict,
- TagTuple,
- PassList,
- TagBytes,
- TagMarkup,
- TagUUID,
- TagDateTime,
- ]
- def __init__(self) -> None:
- self.tags: t.Dict[str, JSONTag] = {}
- self.order: t.List[JSONTag] = []
- for cls in self.default_tags:
- self.register(cls)
- def register(
- self,
- tag_class: t.Type[JSONTag],
- force: bool = False,
- index: t.Optional[int] = None,
- ) -> None:
- """Register a new tag with this serializer.
- :param tag_class: tag class to register. Will be instantiated with this
- serializer instance.
- :param force: overwrite an existing tag. If false (default), a
- :exc:`KeyError` is raised.
- :param index: index to insert the new tag in the tag order. Useful when
- the new tag is a special case of an existing tag. If ``None``
- (default), the tag is appended to the end of the order.
- :raise KeyError: if the tag key is already registered and ``force`` is
- not true.
- """
- tag = tag_class(self)
- key = tag.key
- if key is not None:
- if not force and key in self.tags:
- raise KeyError(f"Tag '{key}' is already registered.")
- self.tags[key] = tag
- if index is None:
- self.order.append(tag)
- else:
- self.order.insert(index, tag)
- def tag(self, value: t.Any) -> t.Dict[str, t.Any]:
- """Convert a value to a tagged representation if necessary."""
- for tag in self.order:
- if tag.check(value):
- return tag.tag(value)
- return value
- def untag(self, value: t.Dict[str, t.Any]) -> t.Any:
- """Convert a tagged representation back to the original type."""
- if len(value) != 1:
- return value
- key = next(iter(value))
- if key not in self.tags:
- return value
- return self.tags[key].to_python(value[key])
- def dumps(self, value: t.Any) -> str:
- """Tag the value and dump it to a compact JSON string."""
- return dumps(self.tag(value), separators=(",", ":"))
- def loads(self, value: str) -> t.Any:
- """Load data from a JSON string and deserialized any tagged objects."""
- return loads(value, object_hook=self.untag)
|